<?php
/**
 * Created by PhpStorm.
 * User: guokunsheng
 * Date: 2019/5/6
 * Time: 11:58
 */

namespace app\common\lib;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use think\Config;
use think\Exception;
use think\Log as ThinkLog;

/**
 * RabbitMQ
 * Class Amqp
 * @package app\common\lib
 */
class AMQP
{
    /**
     * 是否使用阿里云AMQP
     * @var bool
     */
    protected $isAli = true;
    /**
     * AMQP连接
     * @var AMQPStreamConnection|null
     */
    protected $handler = null;
    /**
     * 会话信道
     * @var null
     */
    protected $channel = null;
    /**
     * 配置参数
     * @var array
     */
    protected static $options = [
        'host' => 'localhost',// 连接地址
        'port' => 5672,// 连接端口号
        'username' => '',// 自建使用username
        'password' => '',// 自建使用password
        'access_id' => '',// 子账号 AccessKey
        'access_key' => '',// 子账号 AccessKeySecret
        'uid' => 1919834008307914,// 阿里云主账号
        'vhost' => 'dev',// Vhost
    ];

    /**
     * AMQP 初始化连接
     * AMQP constructor.
     * @throws Exception
     */
    public function __construct()
    {
        /**
         * 若安装amqp扩展，则需要检测扩展
         */
//        if (!extension_loaded('amqp'))
//        {
//            throw new \BadFunctionCallException('not support: amqp');
//        }
        $options = Config::get('rabbitmq');

        self::$options = array_merge(self::$options, $options);

        try
        {
            /**
             * 若使用阿里云AMQP
             */
            if ($this->isAli === true)
            {
                $host = $options['ali_host'];
                $username = $this->getUser();
                $password = $this->getPassword();
                $vhost = $options['vhost'];
            } /**
             * 使用自建AMQP
             */
            else
            {
                $host = $options['host'];
                $username = $options['username'];
                $password = $options['password'];
                $vhost = '/';
            }

            /**
             * 创建连接
             */
            $this->handler = new AMQPStreamConnection($host, $options['port'], $username, $password, $vhost, false, 'AMQPLAIN', null, 'en_US', 30, 120, null, false, 60);
        }
        catch (\Exception $e)
        {
            ThinkLog::write('AMQP：' . $e->getMessage(), 'error');
            throw new Exception('服务异常', 0);
        }
    }

    /**
     * 使用阿里云AMQP时，username处理
     * @return string
     */
    private function getUser()
    {
        $t = '0:' . self::$options['uid'] . ':' . self::$options['access_id'];
        return base64_encode($t);
    }

    /**
     * 使用阿里云AMQP时，password处理
     * @return string
     */
    private function getPassword()
    {
        $ts = time();
        $value = utf8_encode(self::$options['access_key']);
        $key = utf8_encode((string)$ts);
        $sig = strtoupper(hash_hmac('sha1', $value, $key, false));
        return base64_encode(utf8_encode($sig . ':' . $ts));
    }

    /**
     * 开始会话
     */
    private function channel()
    {
        return $this->handler->channel();
    }

    /**
     * 结束当前会话
     * @param object $channel 会话对象
     */
    private function channekClose($channel)
    {
        if (!is_null($channel))
        {
            $channel->close();
        }
    }

    /**
     * 关闭与AMQP的连接
     */
    public function handlerClose()
    {
        if (!is_null($this->handler))
        {
            $this->handler->close();
        }
    }

    /**
     * 结束当前会话 并关闭连接
     * @param object $channel 会话对象
     */
    private function channekHandlerClose($channel)
    {
        if (!is_null($channel))
        {
            $this->channekClose($channel);
        }
        if (!is_null($this->handler))
        {
            $this->handlerClose();
        }
    }

    /**
     * 声明需要使用的队列queue,若没有则自动创建
     * @param object $channel
     * @param string $queue 队列名称
     * @param bool $durable
     */
    private function queueDeclare($channel, $queue = '', $durable = false)
    {
        /**
         * @param string $queue // should be unique in fanout exchange. Let RabbitMQ create a queue name for us
         * @param bool $passive : false  // don't check if a queue with the same name exists
         * @param bool $durable : false  // the queue will not survive server restarts 是否持久化
         * @param bool $exclusive : true // the queue can not be accessed by other channels
         * @param bool $auto_delete : true // the queue will be deleted once the channel is closed.
         */
        $channel->queue_declare($queue, false, $durable, false, false);
    }

    /**
     * 发送消息
     * @param object $channel
     * @param string $msg
     * @param string $exchange
     * @param string $routing_key
     */
    private function basicPublish($channel, $msg, $exchange, $routing_key)
    {
        /**
         * @param string $msg
         * @param string exchange
         * @param string routing_key
         * @param bool mandatory: false
         * @param bool immediate: false
         * @param int ticket
         */
        $channel->basic_publish($msg, $exchange, $routing_key, false, false, null);
    }

    /**
     * 声明需要使用的交换机,若没有则自动创建，指定类型为 direct
     * @param object $channel
     * @param string $exchange
     * @param string $type 'direct'
     *      fanout：该类型路由规则非常简单，会把所有发送到该 Exchange 的消息路由到所有与它绑定的Queue中，相当于广播功能
     *      direct：该类型路由规则会将消息路由到 Binding key 与 Routing key 完全匹配的Queue中
     *      topic：该类型与direct类型相似，只是规则没有那么严格，可以模糊匹配和多条件匹配，即该类型 Exchange 使用 Routing key 模式匹配和字符串比较的方式将消息路由至绑定的Queue
     */
    private function exchangeDeclare($channel, $exchange, $type = 'direct')
    {
        /**
         * @param string $exchange
         * @param string $type
         * @param bool $passive : false // don't check if an exchange with the same name exists
         * @param bool $durable : false // the exchange won't survive server restarts
         * @param bool $auto_delete : true // the exchange will be deleted once the channel is closed.
         */
        $channel->exchange_declare($exchange, $type, false, false, false);
    }

    /**
     * 注册为XX队列的消费者
     * @param object $channel
     * @param string $queue
     * @param callback $callback
     * @param bool $no_ack 是否进行确认消息
     */
    private function basicConsume($channel, $queue, $callback, $no_ack = true)
    {
        /**
         * @param string $queue : Queue from where to get the messages
         * @param string $consumer_tag : Consumer identifier
         * @param bool $no_local : Don't receive messages published by this consumer.
         * @param bool $no_ack : If set to true, automatic acknowledgement mode will be used by this consumer. See https://www.rabbitmq.com/confirms.html for details.
         * @param bool $exclusive : Request exclusive consumer access, meaning only this consumer can access the queue
         * @param bool $nowait : don't wait for a server response. In case of error the server will raise a channel
         * exception
         * @param callback $callback : A PHP Callback
         */
        $channel->basic_consume($queue, '', false, $no_ack, false, false, $callback);
    }

    /**
     * 设置同时最多能处理消息的数量
     * @param object $channel
     * @param int $num 处理消息数量
     */
    private function basicQos($channel, $num = 1)
    {
        /**
         * @param int $prefetch_size
         * @param int $prefetch_count
         * @param bool $a_global
         */
        $channel->basic_qos(null, $num, null);
    }

    /**
     * 发送消息
     * @param string $queue 队列
     * @param string $msg 消息json
     * @param string $exchange 交换机
     * @param string $routing_key 路由key
     * @return bool
     */
    public function send($msg = '', $queue = 'message', $exchange = '', $routing_key = 'message')
    {
        try
        {
            // 创建会话channel
            $channel = $this->channel();
            // 声明需要使用的队列queue,若没有则自动创建
            $this->queueDeclare($channel, $queue, false);
            // 消息内容准备
            $msg = new AMQPMessage($msg);
            // 发送消息
            $this->basicPublish($channel, $msg, $exchange, $routing_key);
            // 关闭会话
            $this->channekClose($channel);
        }
        catch (\Exception $e)
        {
            print_r('AMQP:send:' . $e->getMessage() . PHP_EOL);
            ThinkLog::write('AMQP:send:' . $e->getMessage(), 'error');
            return false;
        }
        return true;
    }

    /**
     * 接收消息
     * @param string $queue
     */
    public function receive($queue = 'message')
    {
        try
        {
            // 创建会话channel
            $channel = $this->channel();
            // 声明需要使用的队列queue
            $this->queueDeclare($channel, $queue, false);

            $callback = function ($msg)
            {
                // 对消息进行处理
                print_r(PHP_EOL . date('Y-m-d H:i:s') . ' 收到消息：' . PHP_EOL . $msg->body . PHP_EOL);
                // TODO something

            };
            // 注册为某队列的消费者
            $this->basicConsume($channel, $queue, $callback, true);

            while (count($channel->callbacks))
            {
                $channel->wait();
            }
            // 关闭会话 && 关闭连接
            $this->channekHandlerClose($channel);
        }
        catch (\Exception $e)
        {
            print_r('AMQP:receive:' . $e->getMessage() . PHP_EOL);
            ThinkLog::write('AMQP:receive:' . $e->getMessage(), 'error');
//            return false;
        }
    }

    /**
     * 发布任务消息
     * @param string $msg 消息json
     * @param string $queue 队列
     * @param int $delay
     * @return bool
     */
    public function task($msg = '', $delay = 0, $queue = 'task')
    {
        try
        {
            $exchange = '';
            $routing_key = $queue;
            $channel = $this->channel();
            $this->queueDeclare($channel, $queue, true);
            $msg = new AMQPMessage($msg, [
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT
            ]);
            // 是否需要进行延时
            if ($delay > 0)
            {
                // header头内容准备
                $header = new AMQPTable();
                $header->set('delay', $delay);
                // 消息设置header头
                $msg->set('application_headers', $header);
            }
            $this->basicPublish($channel, $msg, $exchange, $routing_key);
            $this->channekClose($channel);
        }
        catch (\Exception $e)
        {
            print_r('AMQP:task:' . $e->getMessage() . PHP_EOL);
            ThinkLog::write('AMQP:task:' . $e->getMessage(), 'error');
            return false;
        }
        return true;
    }

    /**
     * 接收任务消息
     * @param string $queue 队列
     */
    public function worker($queue = 'task')
    {
        try
        {
            $channel = $this->channel();
            $this->queueDeclare($channel, $queue, true);
            $callback = function ($msg)
            {
                // 对消息进行处理
                print_r(PHP_EOL . date('Y-m-d H:i:s') . ' 收到任务消息：' . PHP_EOL . $msg->body . PHP_EOL);
                // TODO something
                sleep(3);
                // 消息处理完毕
                print_r(date('Y-m-d H:i:s') . ' 任务消息处理完毕：' . PHP_EOL);
                // 确认消息已处理
                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
            };
            $this->basicQos($channel, 1);
            $this->basicConsume($channel, $queue, $callback, false);
            while (count($channel->callbacks))
            {
                $channel->wait();
            }
            $this->channekHandlerClose($channel);
        }
        catch (\Exception $e)
        {
            print_r('AMQP:worker:' . $e->getMessage() . PHP_EOL);
            ThinkLog::write('AMQP:worker:' . $e->getMessage(), 'error');
        }
    }

    /**
     * 发布延时任务消息(自建AMQP 死信+TTL实现)
     * @param string $msg 消息内容体
     * @param string $queue 队列名，延时任务过期后，通过指定交换机到达的队列
     * @return bool
     */
    public function delayTask($msg = '', $queue = 'delay.task')
    {
        $cacheQueue = 'cache.queue';
        $cacheExchange = 'cache.exchange';
        $delayExchange = 'delay.exchange';
        /**
         * fanout：该类型路由规则非常简单，会把所有发送到该 Exchange 的消息路由到所有与它绑定的Queue中，相当于广播功能
         * direct：该类型路由规则会将消息路由到 Binding key 与 Routing key 完全匹配的Queue中
         * topic：该类型与direct类型相似，只是规则没有那么严格，可以模糊匹配和多条件匹配，即该类型 Exchange 使用 Routing key 模式匹配和字符串比较的方式将消息路由至绑定的Queue
         */
        $type = 'direct';
        try
        {
            // 创建会话channel
            $channel = $this->channel();
            // 声明需要使用的交换机,若没有则自动创建，指定类型为 direct
            $channel->exchange_declare($delayExchange, $type, false, false, false);
            // 声明需要使用的交换机,若没有则自动创建，指定类型为 direct
            $channel->exchange_declare($cacheExchange, $type, false, false, false);
            //设置队列的超时时间
            $tale = new AMQPTable();
            // 设置过期后，交给哪个exchange处理
            $tale->set('x-dead-letter-exchange', $delayExchange);
            // 设置过期后，交给哪个exchange处理
            $tale->set('x-dead-letter-routing-key', $delayExchange);
            // 设置过期时间
            $tale->set('x-message-ttl', 5000);
            // 声明需要使用的队列queue,若没有则自动创建
            $channel->queue_declare($cacheQueue, false, true, false, false, false, $tale);
            // 队列queue绑定到交换机
            $channel->queue_bind($cacheQueue, $cacheExchange, $cacheExchange);
            // 声明需要使用的队列queue,若没有则自动创建
            $channel->queue_declare($queue, false, true, false, false, false);
            // 队列queue绑定到交换机
            $channel->queue_bind($queue, $delayExchange, $delayExchange);
            // 消息内容准备
            $msg = new AMQPMessage($msg, [
                'expiration' => 10000,
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
            ]);
            // 发送消息
            $channel->basic_publish($msg, $cacheExchange, $cacheExchange);
            // 关闭会话
            $this->channekClose($channel);
        }
        catch (\Exception $e)
        {
            print_r('AMQP:delay.task:' . $e->getMessage() . PHP_EOL);
            ThinkLog::write('AMQP:delay.task:' . $e->getMessage(), 'error');
            return false;
        }
        return true;
    }

    /**
     * 接收延时任务消息(自建AMQP)
     * @param string $queue 队列名，参数需与delayTask中$queue参数一致
     */
    public function delayWork($queue = 'delay.task')
    {
        $cacheExchange = 'cache.exchange';
        $delayExchange = 'delay.exchange';
        /**
         * fanout：该类型路由规则非常简单，会把所有发送到该 Exchange 的消息路由到所有与它绑定的Queue中，相当于广播功能
         * direct：该类型路由规则会将消息路由到 Binding key 与 Routing key 完全匹配的Queue中
         * topic：该类型与direct类型相似，只是规则没有那么严格，可以模糊匹配和多条件匹配，即该类型 Exchange 使用 Routing key 模式匹配和字符串比较的方式将消息路由至绑定的Queue
         */
        $type = 'direct';
        try
        {
            // 创建会话channel
            $channel = $this->channel();
            // 声明需要使用的交换机,若没有则自动创建，指定类型为 direct
            $channel->exchange_declare($delayExchange, $type, false, false, false);
            // 声明需要使用的交换机,若没有则自动创建，指定类型为 direct
            $channel->exchange_declare($cacheExchange, $type, false, false, false);
            // 声明需要使用的队列queue,若没有则自动创建
            $channel->queue_declare($queue, false, true, false, false, false);
            // 队列queue绑定到交换机
            $channel->queue_bind($queue, $delayExchange, $delayExchange);
            /**
             * 定义回调函数
             * @param $msg
             */
            $callback = function ($msg)
            {
                // 对消息进行处理
                print_r(PHP_EOL . date('Y-m-d H:i:s') . ' 收到延时任务消息：' . PHP_EOL . $msg->body . PHP_EOL);
                // TODO 业务处理逻辑
                sleep(1);

                // 消息处理完毕
                print_r(date('Y-m-d H:i:s') . ' 延时任务消息处理完毕：' . PHP_EOL);
                // 确认消息已处理
                $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
            };
            // 设置同时最多能处理消息的数量 1
            $channel->basic_qos(null, 1, null);
            // 注册为某队列的消费者
            $channel->basic_consume($queue, '', false, false, false, false, $callback);
            while (count($channel->callbacks))
            {
                $channel->wait();
            }
            // 关闭会话 && 关闭连接
            $this->channekHandlerClose($channel);
        }
        catch (\Exception $e)
        {
            print_r('AMQP:delay:worker:' . $e->getMessage() . PHP_EOL);
            ThinkLog::write('AMQP:delay:worker:' . $e->getMessage(), 'error');
//            return false;
        }
    }
}
